#pragma once
#ifndef KAFKACONSUMER_H
#define KAFKACONSUMER_H

#include <iostream>
#include <string>
#include <cstdlib>
#include <cstdio>
#include <csignal>
#include <cstring>
#include <list>
#include <librdkafka/rdkafkacpp.h>
#include <vector>
#include <fstream>
#include <mutex>
#include <functional>

using std::string;
using std::list;
using std::cout;
using std::endl;
using std::vector;
using std::fstream;
using std::mutex;

/**
 *  kafka消费者模型
 * 
 */
class KafkaConsumer{
public:
    KafkaConsumer(const std::string brokers,string topic, std::string groupid, int32_t artition);
    virtual ~KafkaConsumer();

    //初始化
    bool init();
    //开始获取消息
    void start(int timeout_ms);
    //停止
    void stop();
    //订阅主题
    void subscribe(string topic);
    //取消某个主题的订阅
    bool unsubscribe(string topic);

    //给外界注册的事件回调函数
    void sub_msg_function(std::function<void(string)> fun){this->m_msg_fun=fun;}
private:
    void message_consume(RdKafka::Message* message,void* opaque);
    
private:

    string m_strBroker;
    string m_strGroupID;
    int32_t m_partitionSize;
    bool m_isRun=false;

    RdKafka::KafkaConsumer* m_consumerPtr=nullptr;

    //消息接收成功后给外界的回调函数
    std::function<void(string)> m_msg_fun;
    //topics
    vector<string> m_topics;
    //用来保护修改数组的锁
    mutex m_lock;
};



#endif